package com.ikas.ai.server.kafka;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.JSONValidator;
import com.ikas.ai.model.KafkaMeteData;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 工况流程topic消费
 */
@Slf4j
@Component
public class KafkaWorkFlowConsumerService {

    @Autowired
    private WorkInstanceService workInstanceService;


//    @Resource(name = "kafka-work-flow")
//    private ThreadPoolTaskExecutor workFlowExecutor;

    // 消费监听
    @KafkaListener(topics = {KafkaInitialConfiguration.TOPIC_U01}, topicPattern = "0", groupId = KafkaInitialConfiguration.WORK_FLOW_TOPIC_GROUP)
    public void onMessageU01(List<ConsumerRecord<String, String>> list, Acknowledgment ack) {
        processKafkaMessage(list, ack, KafkaInitialConfiguration.TOPIC_U01);
    }

    // 消费监听
    @KafkaListener(topics = {KafkaInitialConfiguration.TOPIC_U02}, topicPattern = "0", groupId = KafkaInitialConfiguration.WORK_FLOW_TOPIC_GROUP)
    public void onMessageU02(List<ConsumerRecord<String, String>> list, Acknowledgment ack) {
        processKafkaMessage(list, ack, KafkaInitialConfiguration.TOPIC_U02);
    }

    //     消费监听
    @KafkaListener(topics = {KafkaInitialConfiguration.TOPIC_U03}, topicPattern = "0", groupId = KafkaInitialConfiguration.WORK_FLOW_TOPIC_GROUP)
    public void onMessageU03(List<ConsumerRecord<String, String>> list, Acknowledgment ack) {
        processKafkaMessage(list, ack, KafkaInitialConfiguration.TOPIC_U03);
    }

    // 消费监听
    @KafkaListener(topics = {KafkaInitialConfiguration.TOPIC_U04}, topicPattern = "0", groupId = KafkaInitialConfiguration.WORK_FLOW_TOPIC_GROUP)
    public void onMessageU04(List<ConsumerRecord<String, String>> list, Acknowledgment ack) {
        processKafkaMessage(list, ack, KafkaInitialConfiguration.TOPIC_U04);
    }

    public void processKafkaMessage(List<ConsumerRecord<String, String>> list, Acknowledgment ack, String machineNo) {
        Instant start = Instant.now();
        AtomicReference<Long> offset = new AtomicReference<>(0L);
        list.forEach(record -> {
            offset.set(record.offset());
            try {
                List<KafkaMeteData> meteDataList = new ArrayList<>();
                final JSONValidator.Type type = JSONValidator.from(record.value()).getType();
                if (type == JSONValidator.Type.Array) {
                    meteDataList = JSONArray.parseArray(record.value(), KafkaMeteData.class);
                } else if (type == JSONValidator.Type.Object) {
                    meteDataList = Collections.singletonList(JSONObject.parseObject(record.value(), KafkaMeteData.class));
                }
                if (CollectionUtil.isEmpty(meteDataList)) {
                    return;
                }
                log.info("===WORKFLOW-KAFKA===ConsumerRecord_list:size:{},,patition:{},offset:{}", list.size(), record.partition(), record.offset());
                //设置类型排序
                meteDataList.stream().forEach(kafkaMeteData -> {
                    log.info("offset:{},kafkaMeteData:{}", offset, JSONUtil.toJsonStr(kafkaMeteData));
                    workInstanceService.processKafkaMeteData(kafkaMeteData, record.offset());
                });
                log.info("===WORKFLOW-KAFKA===>offset>[{}]-END-timestamp>[{}]", record.offset()
                        , record.timestamp());
            } catch (Exception e) {
                log.error("=====测点数据处理错误======", e);
            }
        });
        log.info("offset:{},开始kafka提交", offset);
        //手动提交
        ack.acknowledge();
        log.info("offset:{},开始kafka提交完成", offset);
        Instant end = Instant.now();
        log.info("machineNo:{},工况kafka消费执行时长毫秒：{},offset:{}", machineNo, Duration.between(start, end).toMillis(), offset);

    }

}



















